package com.taiji.dicom4cloud.smart.rfid.mqtt;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.taiji.dicom4cloud.smart.rfid.manager.IMessageStrategyManagerService;
import com.taiji.dicom4cloud.smart.rfid.model.RfidMessage;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;

import javax.annotation.Resource;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Configuration
public class MqttConfig {

    private static final Logger LOG = LoggerFactory.getLogger(MqttConfig.class);


    @Resource(name = "iRfidMessageStrategyManagerService")
    private IMessageStrategyManagerService iMessageStrategyManagerService;


    public static final String CHANNEL_NAME_IN = "mqttInputChannel";

    public static final String CHANNEL_NAME_OUT = "mqttOutputChannel";

    @Value("${spring.mqtt.url}")
    private String hostUrl;

    @Value("${spring.mqtt.client.id}")
    private String clientId;

    @Value("${spring.mqtt.username}")
    private String username;

    @Value("${spring.mqtt.password}")
    private String password;

    @Value("${spring.mqtt.keepAliveInterval}")
    private int keepAliveInterval;

    @Value("${spring.mqtt.default.topic}")
    private String defaultTopic;

    ExecutorService fixedThreadPool = Executors.newFixedThreadPool(10);

    @Bean
    public MqttConnectOptions getMqttConnectOptions() {
        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
        mqttConnectOptions.setUserName(username);
        mqttConnectOptions.setPassword(password.toCharArray());
        mqttConnectOptions.setServerURIs(new String[]{hostUrl});
        mqttConnectOptions.setKeepAliveInterval(keepAliveInterval);
        mqttConnectOptions.setMaxInflight(1000);
        return mqttConnectOptions;
    }

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(getMqttConnectOptions());
        return factory;
    }

    @Primary
    @Bean(name = CHANNEL_NAME_IN)
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    // Consumer
    @Bean
    public MessageProducer inbound(@Qualifier(CHANNEL_NAME_IN) MessageChannel messageChannel) {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(clientId + "_inbound", mqttClientFactory(), new String[]{defaultTopic});
        adapter.setCompletionTimeout(5000);
        DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter();
        converter.setPayloadAsBytes(true);
        adapter.setConverter(converter);
        adapter.setQos(0);
        adapter.setOutputChannel(messageChannel);
        return adapter;
    }

    @Bean
    @ServiceActivator(inputChannel = CHANNEL_NAME_IN)
    public MessageHandler handler() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                String topic = message.getHeaders().get("mqtt_receivedTopic").toString();
                //byte[] payload = (byte[]) message.getPayload();
                String payload = new String((byte[]) message.getPayload());
                dealSubscription(topic, payload);
            }
        };
    }

    private void dealSubscription(String topic, String payload) {
        fixedThreadPool.execute(new Runnable() {
            @Override
            public void run() {
                LOG.info("topic={} && payload={}", topic, payload);
                synchronized (this){
                    JSONObject data = JSONObject.parseObject(payload);
                    String clientId = data.getString("clientId");
                    String command = data.getString("command");

                    RfidMessage message = new RfidMessage();
                    message.setClientId(clientId);
                    message.setCommand(command);
                    message.setData(data.get("data"));


               /*     String clientId = "X1234";
                    String command = "inventory-data";
                    JSONArray list = new JSONArray();
                    list.add("1111111111");
                    message.setClientId(clientId);
                    message.setCommand(command);
                    message.setData(list);*/

                    iMessageStrategyManagerService.handle(message);
                }


            }
        });
    }

    // publisher
    @Bean(name = CHANNEL_NAME_OUT)
    public MessageChannel mqttOutputChannel() {
        return new DirectChannel();
    }

    @Bean
    @ServiceActivator(inputChannel = CHANNEL_NAME_OUT)
    public MessageHandler outbound() {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId + "_outbound", mqttClientFactory());
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic(defaultTopic);
        return messageHandler;
    }
}
